package memory

import (
	"context"
	"fmt"
	"sync"

	"github.com/louloulin/dataflare/pkg/core"
	"github.com/louloulin/dataflare/pkg/messages"
)

// MemorySinkConnector 是一个内存接收器连接器
type MemorySinkConnector struct {
	// records 是写入的记录
	records []messages.Record
	// mu 是保护记录的互斥锁
	mu sync.RWMutex
	// running 表示连接器是否正在运行
	running bool
}

// NewMemorySinkConnector 创建一个新的内存接收器连接器
func NewMemorySinkConnector() (core.SinkConnector, error) {
	return &MemorySinkConnector{}, nil
}

// Init 初始化连接器
func (c *MemorySinkConnector) Init(ctx context.Context, config map[string]interface{}) error {
	c.records = make([]messages.Record, 0)
	return nil
}

// Start 启动连接器
func (c *MemorySinkConnector) Start(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	c.running = true
	return nil
}

// Stop 停止连接器
func (c *MemorySinkConnector) Stop(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	c.running = false
	return nil
}

// Write 向接收器写入记录
func (c *MemorySinkConnector) Write(ctx context.Context, record messages.Record) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// 检查是否正在运行
	if !c.running {
		return fmt.Errorf("connector not running")
	}

	// 添加记录
	c.records = append(c.records, record)
	return nil
}

// Flush 刷新缓冲记录
func (c *MemorySinkConnector) Flush(ctx context.Context) error {
	// 没有缓冲，所以不需要刷新
	return nil
}

// GetRecords 获取写入的记录
func (c *MemorySinkConnector) GetRecords() []messages.Record {
	c.mu.RLock()
	defer c.mu.RUnlock()

	// 创建记录的副本
	records := make([]messages.Record, len(c.records))
	copy(records, c.records)
	return records
}

// 注册连接器
func init() {
	// 注册内存接收器连接器
	core.RegisterSinkConnector("memory", NewMemorySinkConnector)
}
